package cn.tedu.csmall.cart.webapi.kafka;

import cn.tedu.csmall.commons.pojo.cart.model.Cart;
import com.google.gson.Gson;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

// 要接收kafka的消息需要讲接收消息的对象保存到Spring容器中
// 因为KafkaTemplate是spring在管理的
@Component
public class Consumer {

    // SpringKafka接收消息依靠了框架提供的"监听机制"
    // 框架中有一个线程,一直实时关注kafka的消息情况
    // 如果我们指定的话题名称(myCart)接收了消息,那么这条线程就会自动调用下面的方法
    @KafkaListener(topics = "myCart")
    // 下面定义的方法就是接收到消息后运行的方法
    // 这个方法有参数,参数类型是固定的,参数的值就是监听器接收到的消息内容
    public void received(ConsumerRecord<String,String> record){
        // 参数类型必须是ConsumerRecord
        // 泛型<[话题名称的类型],[消息的类型]>
        // 我们可以将record视为从kafka中接收到的消息对象
        String json=record.value();
        // json可能的值: {"id":2,"commodityCode":"PC100","price":74,"count":8,"userId":"UU100"}
        Gson gson=new Gson();
        // gson也可以将json字符串转换为java对象
        Cart cart=gson.fromJson(json,Cart.class);
        System.out.println("接收到了消息:"+cart);
    }

}
